package com.jeffse.light.stream.app.function;

import com.jeffse.light.stream.app.pojo.AvgTemperature;
import com.jeffse.light.stream.app.pojo.Temperature;
import com.jeffse.light.stream.core.data.context.KeyedContext;
import com.jeffse.light.stream.core.data.context.ProcessContext;
import com.jeffse.light.stream.core.data.descriptor.stateDescriptor.ListStateDescriptor;
import com.jeffse.light.stream.core.data.descriptor.stateDescriptor.ValueStateDescriptor;
import com.jeffse.light.stream.core.data.state.ListState;
import com.jeffse.light.stream.core.data.state.ValueState;
import com.jeffse.light.stream.core.function.base.KeyedProcessFunction;

import java.util.ArrayList;
import java.util.List;
import java.util.stream.Collectors;

public class OutPutSideFunction extends KeyedProcessFunction<String, Temperature, AvgTemperature> {
    private ValueState<Long> timestamp;
    private ValueState<Boolean> isTrigger;
    private ListState<Temperature> listState;

    @Override
    public void open() {
        super.open();
        timestamp = getRuntimeContext().getState(new ValueStateDescriptor("timestamp", this));
        isTrigger = getRuntimeContext().getState(new ValueStateDescriptor("isTrigger", this));
        listState = getRuntimeContext().getListState(new ListStateDescriptor("listState", this));
    }


    @Override
    public void processElement(ProcessContext context, Temperature value) throws Exception {
        if(timestamp.value() == null){
            Long time = (value).getTimestamp() + 5000;
            getTimerService().registerProcessingTimeTimer(time);
            timestamp.update(time);
        }
        List<Temperature> temperatureList = new ArrayList<>();
        if(listState.getList() != null){
            temperatureList.addAll(listState.getList());
        }

        temperatureList.add(value);
        Long lastTime = (value).getTimestamp() - 5000;
        List<Temperature> temperatureList1 = temperatureList.stream().filter(item-> item.getTimestamp() > lastTime).collect(Collectors.toList());
        listState.update(temperatureList1);

        if(isTrigger.value() != null){
            compute();
        }
    }

    @Override
    public void onTimer(long timestamp, KeyedContext ctx, ProcessContext sourceContext) {
        isTrigger.update(true);
        compute();
    }

    public void compute() {
        Double result = listState.getList().stream().mapToDouble(value -> value.getValue()).average().getAsDouble();
        AvgTemperature avgTemperature = new AvgTemperature((String) getCurrentKey(), result, System.currentTimeMillis());
        //log.error("计算出的结果" + avgTemperature.toString() + "=====================================" +  listState.getList().size());
        try {
            collect(avgTemperature);
        }catch (Exception e){

        }

    }
}
